package arima;

import bean.DataPoint;
import bean.Point;
import com.cloudera.sparkts.models.ARIMA;
import com.cloudera.sparkts.models.ARIMAModel;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.spark.mllib.linalg.DenseVector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.ScanParams;
import tools.RedisTool;
import tools.ZookeeperClient;
import util.RegressionLine;
import util.TsUtils;

import java.io.IOException;
import java.time.LocalDate;
import java.util.*;
import java.util.regex.Pattern;

/**
 * @author :chensi
 */
public class ArimaForecast {
    public static void main(String[] args) throws Exception {
        if (args.length < 3 || "--help".equals(args[0]) || "-h".equals(args[0])) {
            System.out.println("Usage:\r\nhadoop jar <jarpath> <inputFilePath> <zookeeperAddress> <reduceNum>");
            System.exit(1);
        }
        Configuration conf = new Configuration();
        //acquire data from redis
        ZookeeperClient zkClient = new ZookeeperClient();
        String inputPath = args[0];
        String zkAddr = args[1];
        String reduceNum = args[2];
        String outputPath = "/tmp/"+System.currentTimeMillis();
        ZooKeeper zk = zkClient.getZk(zkAddr, 30000);
        Properties redisProp = zkClient.getAll(zk, "/conf_htiiot/redis");
        //save config of redis
        conf.set("redis.address", redisProp.getProperty("address"));
        conf.set("redis.port", redisProp.getProperty("port"));
        conf.set("redis.auth", redisProp.getProperty("auth"));
        //clear useless key in redis
        Jedis jedis = new Jedis(redisProp.getProperty("address"), Integer.parseInt(redisProp.getProperty("port")));
        jedis.auth(redisProp.getProperty("auth"));
        removeInvalidKeys(jedis);
        jedis.close();
        //create job
        Job job = Job.getInstance(conf,"ArimaPredict");
        FileSystem fs = FileSystem.get(conf);
        FileStatus[] fileStatuses = fs.listStatus(new Path(inputPath));
        Path[] paths = FileUtil.stat2Paths(fileStatuses);
        job.setJarByClass(ArimaForecast.class);
        FileInputFormat.setInputPaths(job, paths);
        LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
        FileOutputFormat.setOutputPath(job,new Path(outputPath));
        job.setInputFormatClass(TextInputFormat.class);
        job.setMapperClass(ArimaMap.class);
        job.setReducerClass(ArimaReduce.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DataPoint.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(NullWritable.class);
        job.setNumReduceTasks(Integer.parseInt(reduceNum));
        boolean state = job.waitForCompletion(true);
        fs.delete(new Path(outputPath),true);
        fs.close();
        System.exit(state == true ? 0 : 1);
    }

    public static  Pattern timeP = Pattern.compile("^[1-9]\\d{3}-(0[1-9]|1[0-2])-(0[1-9]|[1-2][0-9]|3[0-1])$");
    /**
     * 删除今天以前的key
     *
     * @param jedis
     */
    private static void removeInvalidKeys(Jedis jedis) {
        List<String> keyList = RedisTool.scanKeys(jedis, new ScanParams().match("arima-*").count(100000));
        List<String> invalidKeys = new ArrayList<>();
        String today = LocalDate.now().toString();
        for (String key : keyList) {
            String[] splits = key.split("arima-");
            if(splits.length>1) {
                if(timeP.matcher(splits[1]).matches()) {
                    if (splits[1].compareTo(today) == -1) {
                        invalidKeys.add(key);
                    }
                }
            }
        }
        if(invalidKeys.size()!=0) {
            jedis.del(invalidKeys.toArray(new String[invalidKeys.size()]));
        }
    }

    static class ArimaMap extends Mapper<LongWritable, Text, Text, DataPoint> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            InputSplit inputSplit = context.getInputSplit();
            String[] splits = ((FileSplit) inputSplit).getPath().toString().split("/");
            String time = splits[splits.length - 2];
            String[] times = time.split("-");
            long ts = Long.parseLong(times[0]) * 10000 + Long.parseLong(times[1]) * 100 + Long.parseLong(times[2]);
            String[] lines = value.toString().split("\\t");
            //reverse dn to avoid data skew
            String reverseDn = StringUtils.reverse(lines[0]);
            //The DataPoint`s value here present the degradation time not the metric value
            //And the DataPoit`s time present the date like 20181204 not timestamp
            context.write(new Text(reverseDn), new DataPoint(lines[0], ts, Double.parseDouble(lines[1])));
        }
    }

    static class ArimaReduce extends Reducer<Text, DataPoint, NullWritable, NullWritable> {

        private static org.slf4j.Logger logger = LoggerFactory.getLogger(ArimaReduce.class);
        private static final LocalDate YESDAY = LocalDate.now().minusDays(1);
        public static final int YESDAYVALUE = YESDAY.getYear() * 10000 + YESDAY.getMonth().getValue() * 100 + YESDAY.getDayOfMonth();

        /**
         * calculate number of days difference
         * day2 - day1
         * day format like:20181204
         *
         * @param day1
         * @param day2
         * @return
         */
        public long diffDays(int day1, int day2) {
            LocalDate begin = LocalDate.of(day1 / 10000, day1 / 100 % 100, day1 % 100);
            LocalDate end = LocalDate.of(day2 / 10000, day2 / 100 % 100, day2 % 100);
            return end.toEpochDay() - begin.toEpochDay();
        }

        private Jedis jedis = null;
        private String redisForcastKey = "arima-";
        private int daysToForecast = 8;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            String address = context.getConfiguration().get("redis.address");
            String port = context.getConfiguration().get("redis.port");
            String auth = context.getConfiguration().get("redis.auth");
            jedis = new Jedis(address, Integer.parseInt(port));
            jedis.auth(auth);
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            if (jedis != null) {
                jedis.close();
            }
        }

        @Override
        protected void reduce(Text dn, Iterable<DataPoint> values, Context context) throws IOException, InterruptedException {
            List<DataPoint> data = new ArrayList<>();
            Iterator<DataPoint> it = values.iterator();
            while (it.hasNext()) {
                DataPoint dp = it.next();
                data.add(new DataPoint(dp.getDn(), dp.getTs(), dp.getValue()));
            }
            data.sort((o1, o2) -> (int) (o1.getTs() - o2.getTs()));
            //if data size over than 6 then to forecast
            if(data.size()>6) {
                //Use ARIMA to forecast
                double[] dayArr = TsUtils.generateDayArr(data);
                if (dayArr != null) {
                    DenseVector dayVector = TsUtils.fillLinear(Vectors.dense(dayArr));
                    int daysToForecastAct = daysToForecast;
                    if (data.get(data.size() - 1).getTs() != YESDAYVALUE) {
                        daysToForecastAct = daysToForecastAct + (int) diffDays((int) data.get(data.size() - 1).getTs(), YESDAYVALUE);
                    }
                    ARIMAModel arimaModel = ARIMA.autoFit(dayVector, 5, 2, 5);
                    double[] forecast = TsUtils.arimaForecast(arimaModel, dayVector, daysToForecastAct, daysToForecast);
                    if (forecast == null) {
                        //use linear
                        logger.debug("Using linear to forecast!DN: {} DayVector: {}",StringUtils.reverse(dn.toString()),Arrays.toString(dayVector.toArray()));
                        RegressionLine line = new RegressionLine();
                        int i = 1;
                        for (double v : dayVector.toArray()) {
                            line.addPoint(new Point(i++, v));
                        }
                        forecast = TsUtils.linearForeast(line, daysToForecastAct, i, daysToForecast);
                    }
                    //presist forecast
                    if (forecast != null && forecast.length == daysToForecast) {
                        forecast = removeZeroValue(forecast);
                        LocalDate today = YESDAY.plusDays(1);
                        for (int i = 0; i < forecast.length; i++) {
                            String key = redisForcastKey + today.plusDays(i).toString();
                            jedis.hset(key, StringUtils.reverse(dn.toString()), String.valueOf(forecast[i]));
                        }
                    } else {
                        System.err.println("DN: " + dn.toString() + " Forecast ERROR!!!");
                    }
                }
            }
        }

        public double[] removeZeroValue(double[] forecast){
            for(int i=0;i<forecast.length;i++){
                if(forecast[i]<0){
                    forecast[i] = 0;
                }
            }
            return forecast;
        }
    }
}
